package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.func.BeanToJsonStrMapFcuntion;
import com.atguigu.gmall.realtime.app.func.DimAsyncFunction;
import com.atguigu.gmall.realtime.beans.TradeSkuOrderBean;
import com.atguigu.gmall.realtime.utils.DateFormatUtil;
import com.atguigu.gmall.realtime.utils.DorisUtil;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.util.concurrent.TimeUnit;

/**
 * @author Felix
 * @date 2023/10/13
 * sku粒度下单聚合统计
 * 维度：sku
 * 度量：原始金额、优惠券减免金额、活动减免金额、实付金额
 * 需要启动的进程
 *      zk、kafka、maxwell、hdfs、hbase、redis、doris、DwdTradeOrderDetail、DwsTradeSkuOrderWindow
 * 开发流程
 *      准备环境
 *      检查点设置
 *      从kafka的下单事实表主题中读取数据
 *      过滤空消息并将jsonStr转换为jsonObj
 *      去重
 *          为什么会产生重复数据？
 *              因为我们从dwd的下单事实表中读取数据，下单事实表示由业务系统的4张表组成的。
 *              订单明细表、订单表、订单明细活动表、订单明细优惠券表，这4张表进行关联的时候，主表是订单明细，和订单表
 *              进行关联使用的是内连接，和明细活动以及明细优惠券关联的时候使用的是左外连接。
 *              如果使用左外连接，当左表数据先到，右表数据后到的话，会产生如下数据
 *              左表      null    +I
 *              左表      null    -D
 *              左表      右表     +I
 *              这样的数据发送到kafka主题中，kakfa主题会接收到3条消息
 *              左表      null
 *              null
 *              左表      右表
 *              所以当我们从kafka主题中读取数据的时候，需要过滤空以及去重
 *          去重方式1：状态 + 定时器
 *              当第一条数据进来的时候，将数据放到状态中，并注册一个5s后执行的定时器，如果有重复数据，
 *              当重复数据进来的时候，会用当前数据的聚合时间和状态中存在的数据的聚合时间进行比较，如果
 *              当前数据的时间大，将当前数据放到状态中。
 *              等定时器被触发的时候，将状态中的对象传递到下游
 *              不足：不论是否有重复数据，都需要等5s后才会向下游传递数据，时效性较差
 *          去重方式2：状态 + 抵消
 *              当数据来到的时候，将数据放到状态中，并直接传递到下游
 *              如果有重复数据，会将状态中的数据影响到度量的属性进行取反，再向下游传递
 *              将当前数据也要传递到下游
 *              不足：如果有重复了，会传递3条数据到下游，传输量变大了
 *      再次对流中数据进行类型转换   jsonObj->实体类对象(维度、度量)
 *      指定Watermark以及提取事件时间字段  ts*1000
 *      按照统计的维度skuid分组
 *      开窗
 *      聚合
 *      关联维度
 *          维度基本关联
 *              HbaseUtil----JSONObject getDimInfoFromHbase()
 *          优化1：旁路缓存
 *              思路：先从缓存中获取维度数据，如果在缓存中找到了要关联的维度，直接将其作为方法的返回值进行返回(缓存命中)
 *                   如果在缓存中没有找到要关联的维度，发送请求到hbase中查询维度数据，并将查询结果放到缓存中缓存
 *                   起来，方便下次查询使用
 *              缓存产品选型：
 *                  状态：     性能很好，维护性差
 *                  Redis：   性能不错，维护性好 √
 *                  关于Redis的一些设置：
 *                      key:    维度表名:主键值
 *                      type:   string
 *                      expire: 1day
 *                      注意：如果维度数据发生了变化，将缓存的数据进行删除
 *              DimUtil---JSONObject getDimInfo()
 *          优化2：异步IO
 *              为什么要使用异步IO？
 *                  因为在关联维度的时候，如果使用map算子，底层是同步的处理方式，处理完一个元素后再处理下一个元素，
 *                  等待时间比较长，影响时效性。
 *                  可以通过改变并行度提升处理能力，但是更大的并行度意味需要更多的资源，不可能无限制的提升
 *                  所以使用异步的处理方式
 *              Flink如何发送异步请求
 *                  AsyncDataStream.[un]orderWait(
 *                      流,
 *                      如何发送异步请求,implements AsyncFunction,
 *                      超时时间,
 *                      时间单位
 *                  )
 *              Redis以及Hbase支持异步操作
 *                  Redis:StatefulRedisConnection
 *                      RedisUtil
 *                          获取异步连接
 *                          关闭异步连接
 *                          以异步的方式从Redis中读取数据
 *                          以异步的方式向Redis写数据
 *                  Hbase:AsyncConnection
 *                      HbaseUtil
 *                          获取异步连接
 *                          关闭异步连接
 *                          以异步的方式从Hbase中查询维度数据
 *          模板方式设计模式：在父类中定义完成某一个功能的核心算法骨架(步骤)，但是有些步骤延迟到子类中去完成，
 *                      在不改变父类核心算法骨架(步骤)的前提下，每一个子类都可以有自己不同的实现
 *          class DimAsyncFunction extends RichAsyncFunction implements DimJoinFunction{
 *              asyncInvoke{
 *                  //发送异步请求进行维度关联
 *                  CompletableFuture.supplyAsync：创建异步编排对象，执行并行任务，有返回值
 *                      发送异步请求从Redis中获取维度数据
 *                  thenApplyAsync：执行并行任务，有参数，有返回值，会将上一个并行任务的返回结果作为当前任务的参数
 *                      缓存命中：直接输出从redis查的数据
 *                      缓存未命中：发送异步请求到Hbase中查询维度，并将结果放到Redis中
 *                  thenAcceptAsync：执行并行任务，有参数，无返回值
 *                      维度关联：将维度对象的属性补充到流中的对象上-join
 *                      将补充完维度属性的对象 交给resultFuture.complete(集合(obj))
 *              }
 *          }
 *      将关联以及聚合的结果写到Doris中
 */
public class DwsTradeSkuOrderWindow {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);
        //TODO 2.检查点相关的设置
        env.enableCheckpointing(5000L);
        //TODO 3.从kakfa主题中读取数据
        //3.1 声明消费的主题以及消费者组
        String topic = "dwd_trade_order_detail";
        String groupId = "dws_trade_sku_order_group";
        //3.2 创建消费者对象
        KafkaSource<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, groupId);
        //3.3 消费数据封装为流
        DataStreamSource<String> kafkaStrDS
            = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka_source");
        //TODO 4.过滤空消息  并对流中的数据进行类型转换   jsonStr->jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.process(
            new ProcessFunction<String, JSONObject>() {
                @Override
                public void processElement(String jsonStr, Context ctx, Collector<JSONObject> out) throws Exception {
                    if (StringUtils.isNotEmpty(jsonStr)) {
                        JSONObject jsonObj = JSON.parseObject(jsonStr);
                        out.collect(jsonObj);
                    }
                }
            }
        );

        // jsonObjDS.print(">>>>");
        //TODO 5.去重
        //5.1 按照唯一键(订单明细id)进行分组
        KeyedStream<JSONObject, String> orderDetailIdKeyedDS
            = jsonObjDS.keyBy(jsonObj -> jsonObj.getString("id"));
        //5.2 去重方式1：状态 + 定时任务
        /*//不足：每条数据进来后，不论是否重复，都需要等5s后才会传递到下游，影响时效性
        SingleOutputStreamOperator<JSONObject> distinctDS = orderDetailIdKeyedDS.process(
            new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                private ValueState<JSONObject> lastJsonObjState;

                @Override
                public void open(Configuration parameters) throws Exception {
                    ValueStateDescriptor<JSONObject> valueStateDescriptor
                        = new ValueStateDescriptor<JSONObject>("lastJsonObjState", JSONObject.class);
                    lastJsonObjState = getRuntimeContext().getState(valueStateDescriptor);
                }

                @Override
                public void processElement(JSONObject jsonObj, Context ctx, Collector<JSONObject> out) throws Exception {
                    //从状态中获取数据
                    JSONObject lastJsonObj = lastJsonObjState.value();
                    if (lastJsonObj == null) {
                        //没有产生重复数据  将当前这条数据放到状态中  并注册一个5s后执行的定时器
                        lastJsonObjState.update(jsonObj);
                        long currentProcessingTime = ctx.timerService().currentProcessingTime();
                        ctx.timerService().registerProcessingTimeTimer(currentProcessingTime + 5000L);
                    } else {
                        //说明数据重复了，用当前数据的聚合时间和状态中的数据的聚合时间进行比对，将时间大的放到状态中
                        String ts1 = lastJsonObj.getString("聚合时间");
                        String ts2 = jsonObj.getString("聚合时间");
                        if (ts2.compareTo(ts1) >= 0) {
                            //如果当前数据的聚合时间 大于等于状态中的数据的聚合时间，将当前数据放到状态中
                            lastJsonObjState.update(jsonObj);
                        }
                    }
                }

                @Override
                public void onTimer(long timestamp, OnTimerContext ctx, Collector<JSONObject> out) throws Exception {
                    //当定时器被触发执行的时候，将状态中的数据发送到下游
                    JSONObject jsonObj = lastJsonObjState.value();
                    out.collect(jsonObj);
                    //清除状态
                    lastJsonObjState.clear();
                }
            }
        );*/
        //5.3 去重方式2：状态 + 抵消
        //优点：时效性好       不足：如果出现重复，会向下游发送3条数据，传输量变大
        SingleOutputStreamOperator<JSONObject> distinctDS = orderDetailIdKeyedDS.process(
            new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                private ValueState<JSONObject> lastJsonObjState;

                @Override
                public void open(Configuration parameters) throws Exception {
                    ValueStateDescriptor<JSONObject> valueStateDescriptor
                        = new ValueStateDescriptor<JSONObject>("lastJsonObjState", JSONObject.class);
                    valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.seconds(10)).build());
                    lastJsonObjState = getRuntimeContext().getState(valueStateDescriptor);
                }

                @Override
                public void processElement(JSONObject jsonObj, Context ctx, Collector<JSONObject> out) throws Exception {
                    JSONObject lastJsonObj = lastJsonObjState.value();
                    if (lastJsonObj != null) {
                        //如果状态中的数据不为空，说明有重复，将状态中存在的数据影响到度量值的字段进行取反，传递到下游
                        String splitOriginalAmount = lastJsonObj.getString("split_original_amount");
                        String splitCouponAmount = lastJsonObj.getString("split_coupon_amount");
                        String splitActivityAmount = lastJsonObj.getString("split_activity_amount");
                        String splitTotalAmount = lastJsonObj.getString("split_total_amount");
                        lastJsonObj.put("split_original_amount", "-" + splitOriginalAmount);
                        lastJsonObj.put("split_coupon_amount", "-" + splitCouponAmount);
                        lastJsonObj.put("split_activity_amount", "-" + splitActivityAmount);
                        lastJsonObj.put("split_total_amount", "-" + splitTotalAmount);
                        out.collect(lastJsonObj);
                    }
                    lastJsonObjState.update(jsonObj);
                    out.collect(jsonObj);
                }
            }
        );
        //TODO 6.再次转换流中数据类型 jsonObj->统计的实体类对象
        SingleOutputStreamOperator<TradeSkuOrderBean> beanDS = distinctDS.map(
            new MapFunction<JSONObject, TradeSkuOrderBean>() {
                @Override
                public TradeSkuOrderBean map(JSONObject jsonObj) throws Exception {
                    // {"create_time":"2023-10-12 10:04:28","sku_num":"1","activity_rule_id":"5",
                    // "split_original_amount":"9199.0000","split_coupon_amount":"0.0","sku_id":"18",
                    // "date_id":"2023-10-12","user_id":"358","province_id":"30","activity_id":"4",
                    // "sku_name":"TCL板电视","id":"14255384","order_id":"53078","split_activity_amount":"919.9",
                    // "split_total_amount":"8279.1","ts":"1697162668"}
                    String skuId = jsonObj.getString("sku_id");
                    String splitOriginalAmount = jsonObj.getString("split_original_amount");
                    String splitCouponAmount = jsonObj.getString("split_coupon_amount");
                    String splitActivityAmount = jsonObj.getString("split_activity_amount");
                    String splitTotalAmount = jsonObj.getString("split_total_amount");
                    Long ts = jsonObj.getLong("ts") * 1000;
                    TradeSkuOrderBean orderBean = TradeSkuOrderBean.builder()
                        .skuId(skuId)
                        .originalAmount(new BigDecimal(splitOriginalAmount))
                        .couponAmount(new BigDecimal(splitCouponAmount))
                        .activityAmount(new BigDecimal(splitActivityAmount))
                        .orderAmount(new BigDecimal(splitTotalAmount))
                        .ts(ts)
                        .build();
                    return orderBean;
                }
            }
        );
        // beanDS.print(">>>>");
        //TODO 7.指定Watermark以及提取事件时间字段
        SingleOutputStreamOperator<TradeSkuOrderBean> withWatermarkDS = beanDS.assignTimestampsAndWatermarks(
            WatermarkStrategy
                .<TradeSkuOrderBean>forMonotonousTimestamps()
                .withTimestampAssigner(
                    new SerializableTimestampAssigner<TradeSkuOrderBean>() {
                        @Override
                        public long extractTimestamp(TradeSkuOrderBean orderBean, long recordTimestamp) {
                            return orderBean.getTs();
                        }
                    }
                )
        );
        //TODO 8.按照统计的维度sku进行分组
        KeyedStream<TradeSkuOrderBean, String> skuIdKeyedDS
            = withWatermarkDS.keyBy(TradeSkuOrderBean::getSkuId);
        //TODO 9.开窗
        WindowedStream<TradeSkuOrderBean, String, TimeWindow> windowDS
            = skuIdKeyedDS.window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)));
        //TODO 10.聚合计算
        SingleOutputStreamOperator<TradeSkuOrderBean> reduceDS = windowDS.reduce(
            new ReduceFunction<TradeSkuOrderBean>() {
                @Override
                public TradeSkuOrderBean reduce(TradeSkuOrderBean value1, TradeSkuOrderBean value2) throws Exception {
                    value1.setOriginalAmount(value1.getOriginalAmount().add(value2.getOriginalAmount()));
                    value1.setActivityAmount(value1.getActivityAmount().add(value2.getActivityAmount()));
                    value1.setCouponAmount(value1.getCouponAmount().add(value2.getCouponAmount()));
                    value1.setOrderAmount(value1.getOrderAmount().add(value2.getOrderAmount()));
                    return value1;
                }
            },
            new WindowFunction<TradeSkuOrderBean, TradeSkuOrderBean, String, TimeWindow>() {
                @Override
                public void apply(String groupId, TimeWindow window, Iterable<TradeSkuOrderBean> input, Collector<TradeSkuOrderBean> out) throws Exception {
                    String stt = DateFormatUtil.toYmdHms(window.getStart());
                    String edt = DateFormatUtil.toYmdHms(window.getEnd());
                    String curDate = DateFormatUtil.toDate(window.getStart());
                    for (TradeSkuOrderBean orderBean : input) {
                        orderBean.setStt(stt);
                        orderBean.setEdt(edt);
                        orderBean.setCurDate(curDate);
                        out.collect(orderBean);
                    }
                }
            }
        );
        // reduceDS.print(">>>>");
        //TODO 11.关联sku维度
        /*基本维度关联实现
        SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = reduceDS.map(
            new RichMapFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {
                private Connection conn;
                private Jedis jedis;

                @Override
                public void open(Configuration parameters) throws Exception {
                    conn = HbaseUtil.getHbaseConnection();
                    jedis = RedisUtil.getRedisClient();
                }

                @Override
                public void close() throws Exception {
                    HbaseUtil.closeHbaseConnection(conn);
                    RedisUtil.closeRedisClient(jedis);
                }

                @Override
                public TradeSkuOrderBean map(TradeSkuOrderBean orderBean) throws Exception {
                    //根据流中对象获取要关联的维度的主键
                    String skuId = orderBean.getSkuId();
                    //根据维度的主键获取维度对象
                    // id,spu_id,price,sku_name,sku_desc,weight,tm_id,category3_id,sku_default_img,is_sale,create_time
                    // JSONObject dimJsonObj = HbaseUtil.getDimInfoFromHbase(conn, GmallConfig.HBASE_NAMESPACE, "dim_sku_info", skuId);
                    JSONObject dimJsonObj = DimUtil.getDimInfo(jedis, conn, GmallConfig.HBASE_NAMESPACE, "dim_sku_info", skuId);
                    //将维度对象相关的维度属性补充到流中的对象上
                    orderBean.setSkuName(dimJsonObj.getString("sku_name"));
                    orderBean.setSpuId(dimJsonObj.getString("spu_id"));
                    orderBean.setTrademarkId(dimJsonObj.getString("tm_id"));
                    orderBean.setCategory3Id(dimJsonObj.getString("category3_id"));
                    return orderBean;
                }
            }
        );
        // withSkuInfoDS.print(">>>>>");*/
        /*
        抽取模板方法关联维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = reduceDS.map(
            new DimMapFunction<TradeSkuOrderBean>("dim_sku_info") {
                @Override
                public void join(TradeSkuOrderBean orderBean, JSONObject dimJsonObj) {
                    orderBean.setSkuName(dimJsonObj.getString("sku_name"));
                    orderBean.setSpuId(dimJsonObj.getString("spu_id"));
                    orderBean.setTrademarkId(dimJsonObj.getString("tm_id"));
                    orderBean.setCategory3Id(dimJsonObj.getString("category3_id"));
                }

                @Override
                public String getKey(TradeSkuOrderBean orderBean) {
                    return orderBean.getSkuId();
                }
            }
        );

        // withSkuInfoDS.print(">>>>");
        SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = AsyncDataStream.unorderedWait(
            reduceDS,
            new RichAsyncFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {
                private StatefulRedisConnection<String, String> redisConn;
                private AsyncConnection hbaseConn;

                @Override
                public void open(Configuration parameters) throws Exception {
                    redisConn = RedisUtil.getAsyncRedisConnection();
                    hbaseConn = HbaseUtil.getAsyncConnection();
                }

                @Override
                public void close() throws Exception {
                    RedisUtil.closeAysncRedisConnection(redisConn);
                    HbaseUtil.closeAsyncConnection(hbaseConn);
                }

                @Override
                public void asyncInvoke(TradeSkuOrderBean orderBean, ResultFuture<TradeSkuOrderBean> resultFuture) throws Exception {
                    //根据当前流中对象获取要关联的维度主键
                    String skuId = orderBean.getSkuId();
                    //先到Redis中获取维度数据(异步)
                    JSONObject dimJsonObj = RedisUtil.asyncGetDimInfo(redisConn, "dim_sku_info:" + skuId);
                    if (dimJsonObj == null) {
                        //如果在Redis中没有找到要关联的维度，发送请求到Hbase中获取对应的维度数据(异步)
                        dimJsonObj = HbaseUtil.getDimInfoFromHbaseByAsync(hbaseConn, GmallConfig.HBASE_NAMESPACE, "dim_sku_info", skuId);
                        //将habse中查询到的维度数据  放到Redis中缓存(异步)
                        RedisUtil.asyncWriteDim(redisConn, "dim_sku_info:" + skuId, dimJsonObj);
                    }
                    orderBean.setSkuName(dimJsonObj.getString("sku_name"));
                    orderBean.setSpuId(dimJsonObj.getString("spu_id"));
                    orderBean.setTrademarkId(dimJsonObj.getString("tm_id"));
                    orderBean.setCategory3Id(dimJsonObj.getString("category3_id"));

                    resultFuture.complete(Collections.singleton(orderBean));

                }
            },
            60, TimeUnit.SECONDS
        );*/
        SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = AsyncDataStream.unorderedWait(
            reduceDS,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_sku_info") {
                @Override
                public void join(TradeSkuOrderBean orderBean, JSONObject dimJsonObj) {
                    orderBean.setSkuName(dimJsonObj.getString("sku_name"));
                    orderBean.setSpuId(dimJsonObj.getString("spu_id"));
                    orderBean.setTrademarkId(dimJsonObj.getString("tm_id"));
                    orderBean.setCategory3Id(dimJsonObj.getString("category3_id"));
                }

                @Override
                public String getKey(TradeSkuOrderBean orderBean) {
                    return orderBean.getSkuId();
                }

            },
            60, TimeUnit.SECONDS
        );

        // withSkuInfoDS.print(">>>>");
        //TODO 12.关联spu维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withSpuInfoDS = AsyncDataStream.unorderedWait(
            withSkuInfoDS,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_spu_info") {
                @Override
                public void join(TradeSkuOrderBean orderBean, JSONObject dimJsonObj) {
                    orderBean.setSpuName(dimJsonObj.getString("spu_name"));
                }

                @Override
                public String getKey(TradeSkuOrderBean orderBean) {
                    return orderBean.getSpuId();
                }
            },
            60, TimeUnit.SECONDS
        );


        //TODO 13.关联tm维度
        SingleOutputStreamOperator<TradeSkuOrderBean> withTmDS = AsyncDataStream.unorderedWait(
            withSpuInfoDS,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_base_trademark") {
                @Override
                public void join(TradeSkuOrderBean orderBean, JSONObject dimJsonObj) {
                    orderBean.setTrademarkName(dimJsonObj.getString("tm_name"));
                }

                @Override
                public String getKey(TradeSkuOrderBean orderBean) {
                    return orderBean.getTrademarkId();
                }
            },
            60, TimeUnit.SECONDS
        );
        // withTmDS.print(">>>");
        //TODO 14.关联categor3维度
        SingleOutputStreamOperator<TradeSkuOrderBean> c3Stream = AsyncDataStream.unorderedWait(
            withTmDS,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_base_category3") {

                @Override
                public String getKey(TradeSkuOrderBean bean) {
                    return bean.getCategory3Id();
                }

                @Override
                public void join(TradeSkuOrderBean bean,
                                 JSONObject dim) {
                    bean.setCategory3Name(dim.getString("name"));
                    bean.setCategory2Id(dim.getString("category2_id"));
                }
            },
            60,
            TimeUnit.SECONDS
        );

        //TODO 15.关联categor2维度
        SingleOutputStreamOperator<TradeSkuOrderBean> c2Stream = AsyncDataStream.unorderedWait(
            c3Stream,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_base_category2") {

                @Override
                public String getKey(TradeSkuOrderBean bean) {
                    return bean.getCategory2Id();
                }

                @Override
                public void join(TradeSkuOrderBean bean,
                                 JSONObject dim) {
                    bean.setCategory2Name(dim.getString("name"));
                    bean.setCategory1Id(dim.getString("category1_id"));
                }
            },
            60,
            TimeUnit.SECONDS
        );

        //TODO 16.关联categor1维度
        SingleOutputStreamOperator<TradeSkuOrderBean> restulStream = AsyncDataStream.unorderedWait(
            c2Stream,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_base_category1") {

                @Override
                public String getKey(TradeSkuOrderBean bean) {
                    return bean.getCategory1Id();
                }

                @Override
                public void join(TradeSkuOrderBean bean,
                                 JSONObject dim) {
                    bean.setCategory1Name(dim.getString("name"));
                }
            },
            60,
            TimeUnit.SECONDS
        );

        restulStream.print(">>>");
        //TODO 17.将关联的结果写到Doris中
        restulStream
            .map(new BeanToJsonStrMapFcuntion<>())
            .sinkTo(DorisUtil.getDorisSink("dws_trade_sku_order_window"));


        env.execute();
    }
}
